package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.GwmKafkaJson;
import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.function.map.ChargePredictResStandardMapFunction;
import cn.gwm.flink.streaming.function.map.ChargePredictSrcStandardMapFunction;
import cn.gwm.flink.streaming.function.process.ChargePredictProcessFunction;
import cn.gwm.flink.streaming.function.source.ChargePredictMysqlSourceFunction;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.strategy.vehiclepredict.*;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.checkpoint.CheckpointFailureManager;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.*;

/**
 * @Author: GW00280745
 * @Date: 2022/5/10 18:40
 * @Description: 等价 ChargeEndTempDwsTask 充电末端温度预估服务
 * 任务执行方式：
 * mapred queue -list
 * currdir=$(cd $(dirname $0); cd ..;pwd)
 * flink run -d -m yarn-cluster -yqu root.users.flink -p 3 -ys 3 -ynm dws_chargeestimate -yjm 2048 -ytm 2048
 *      \ -c cn.gwm.flink.streaming.task.ChargePredictDwsTask $currdir/StreamingAnalysis-1.0-SNAPSHOT-jar-with-dependencies.jar -p 3
 *
 *       kafka-topics--delete--zookeeper gw04:2181--topic bms_dws_endtemp
 *       kafka-topics --create --zookeeper 10.31.8.113:2181 --replication-factor 1 --partitions 3 --topic dws_multidimension_predict
 *       kafka-topics --describe --zookeeper gw04:2181 --topic bms_dws_endtemp
 *       kafka-console-consumer --bootstrap-server gw04:9092  --topic bms_dws_endtemp
 */
public class ChargePredictDwsTask {


    public static boolean predict_type_1_only_map =true;
    /**
     * @description:  2-多维模型 1-末端温度预测
     **/
    public static int predict_type=2;
    /**
     * @description:  1-内网地址 2-阿里云地址 3-内网验证 4- 本地debug
     *
     * private String cpDir = "hdfs://bd-zcpt-bd-zcpt-test-000002:8020/external/data/dws/dws_multidimensional_charge_cp"
     * private String cpDir = "hdfs://nameservice1/external/data/dws/dws_multidimensional_charge_cp";
     **/
    public static int aliOrInner=3;
    private boolean isTest = false;
    private String cpDir = "hdfs://nameservice1/external/data/dws/dws_multidimensional_charge_cp";
    /**
     * @description:  充电明细主题
     * 测试主题
     * private String dwmTopic = "dws_multidimensional_debug_230410";
     * private String dwmTopic = "bms_charge_dwm_detail";
     **/
    private String dwmTopic = "bms_charge_dwm_detail";
    /**
     * @description:
     * private String dwmTopic = "dwm_charge_detail";
     * private String dwmTopic = "bms_charge_dwm_detail";
     * 多维模型主题
     * private String dwmTopic = "dwm_multidimensional_charge";
     * 郭艳艳充电预测主题
     * private String dwsTopic = "dws_charge_predict";
     * 多维模型充电预测主题
     * private String dwsTopic = "dws_multidimensional_predict";
     * private String dwsTopic = "dws_multidimensional_predict_230201";
     **/
    private String dwsTopic = "dws_multidimensional_predict";

    public static void main(String[] args) throws Exception{
        ChargePredictDwsTask estimateDwsTask = new ChargePredictDwsTask();
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        estimateDwsTask.topicSettings(parameterTool);
        //start. 加载配置(静态配置+动态配置)
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        if(parameterTool.has(estimateDwsTask.innerParamP)){env.setParallelism(parameterTool.getInt(estimateDwsTask.innerParamP,1));}
        estimateDwsTask.checkPoint(env);
        BroadcastStream<JSONObject> dynamicConfInfo = estimateDwsTask.dynamicConfInfo(env);
        //1. 数据源标准化
        DataStream<JSONObject> srcDataStream = estimateDwsTask.sourceStd(env);
        //2. 数据处理
        DataStream<JSONObject> processDataStream = srcDataStream
                .keyBy((KeySelector<JSONObject, Object>) value -> value.getString(ChargePredictFields.ins().srcFieldVin))
                .connect(dynamicConfInfo)
                .process(new ChargePredictProcessFunction())
                ;
        //3. 结果标准化
        DataStream resultData = estimateDwsTask.dataStd(processDataStream);
        //4. 数据输出
        estimateDwsTask.outPutData(resultData);
        //end. 启动
        env.execute(estimateDwsTask.dwsTopic);
    }
    private String innerParamP="p";
    private int innerParam1 = 1;
    private int innerParam2 = 2;
    private String oneSocName = "one_soc_update";
    private void topicSettings(ParameterTool parameterTool) {
//        if (ChargePredictDwsTask.predict_type==1){
//            dwmTopic = "bms_charge_dwm_detail";
//            dwsTopic = "dws_charge_predict";
//        }else {
////            dwmTopic = "dwm_multidimensional_charge";
//            dwmTopic = "bms_charge_dwm_detail";
//            dwsTopic = "dws_multidimensional_predict_0130";
//        }
        // System.getProperty 获取的是 VM options 即系统变量 常见的有 -X -D
        // System.getenv Environment variables没有前缀，优先级低于VM options 例如 password=123456789;name=lisi
//        String boot_active = System.getProperty("boot_active");
        String bootActive = System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE);
        System.out.println(" boot_active ===== "+bootActive);
        System.out.println(" predict_type ===== "+ChargePredictDwsTask.predict_type);
        System.out.println(" only_map ===== "+ChargePredictDwsTask.predict_type_1_only_map);
        System.out.println(" aliOrInner ===== "+ChargePredictDwsTask.aliOrInner);
    }

    /**
     * @description: 结果输出
     * @author:  GW00280745
     **/
    private void outPutData(DataStream resultData) {
        //end. 信息sink到对应topic并持久化
        if (isTest){
            resultData.printToErr();
        }else {
            // kafka-topics --delete --zookeeper gw04:2181 --topic bms_dws_endtemp
            // kafka-topics --create --zookeeper gw04:2181 --replication-factor 1 --partitions 3 --topic bms_dws_endtemp
            // kafka-topics --describe --zookeeper gw04:2181 --topic bms_dws_endtemp
            // kafka-console-consumer --bootstrap-server gw04:9092  --topic bms_dws_endtemp
            if (ChargePredictDwsTask.predict_type==innerParam1){
                sink2Kafka(resultData);
                ChargeData2Hive.builder().input(resultData).build().process();
            }else if (ChargePredictDwsTask.predict_type==innerParam2){
                sink2Kafka(resultData);
//                ChargeData2Hive.builder().input(resultData).build().process();
            }
//            EndTempData2HBase.builder().input(resultData).build().process()
        }
    }
    public void sink2Kafka(DataStream dataStream) {
        String kafkaServer = ChargePredictFields.ins().confInfo.getOrDefault("bootstrap.servers","10.31.8.112:9092,10.31.8.113:9092,10.31.8.114:9092");
        Properties prodProps = new Properties();
        prodProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        prodProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prodProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
//        prodProps.put(ProducerConfig.ACKS_CONFIG,1)
        String topic = dwsTopic;
        if (aliOrInner == DwsStrConstant.ins().num1){
            topic = "dws_multidimensional_predict";
        }else if (aliOrInner == DwsStrConstant.ins().num3){
//            topic = "dws_multidimensional_predict_obc_validation";
            topic = "dws_multidimensional_obc_predict_target";
        }
        System.out.println(" 目的地dwsTopic ===== "+topic);
        ChargePredictProducer producer = new ChargePredictProducer(prodProps, topic);
        dataStream
                . map((MapFunction) value -> {
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put(ChargePredictFields.ins().srcFieldSource, value);
                    return GwmKafkaJson.builder().timestamp(System.currentTimeMillis()).type("json").body(jsonObject).build();
                })
                .returns(Types.GENERIC(JSONObject.class))
                .addSink(producer)
                .name("sink2kafka")
        ;
    }
    /**
     * @description: 结果标准化
     * @author:  GW00280745
     **/
    private DataStream dataStd(DataStream<JSONObject> processDataStream){
        JSONObject resJson = ChargePredictFields.ins().resInfo;
        ChargePredictResStandardMapFunction resStandard = new ChargePredictResStandardMapFunction(resJson);
        SingleOutputStreamOperator returns = processDataStream
                .map(resStandard)
                .returns(Types.GENERIC(JSONObject.class))
                ;
        return returns;
    }
    /**
     * @description: source数据标准化，目前只是简单的字段归一，未涉及字段和value的transform
     * @author:  GW00280745
     * @date: 2022/7/4 8:10
     * @param:
     * @param: env
     * @return:
     * @return: org.apache.flink.streaming.api.datastream.DataStream<com.alibaba.fastjson.JSONObject>
     **/
    private DataStream<JSONObject> sourceStd(StreamExecutionEnvironment env) {
        //1.构造consumer
        String kafkaServer = ChargePredictFields.ins().confInfo.getOrDefault("bootstrap.servers","10.31.8.112:9092,10.31.8.113:9092,10.31.8.114:9092");
//        String kafkaServer  "alikafka-pre-cn-zpr34hg5c004-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zpr34hg5c004-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zpr34hg5c004-3-vpc.alikafka.aliyuncs.com:9092"
        String hHmmss = DateFormatUtils.format(System.currentTimeMillis(), "HHmmss");
        String topic = dwmTopic;
        if (aliOrInner== DwsStrConstant.ins().num1){
            //内网
            topic = "dwm_charge_detail";
        }else if (aliOrInner== DwsStrConstant.ins().num3){
//            topic = "dws_multidimensional_obc_validationset";
            topic = "dws_multidimensional_obc_predict_source";
        }else if (aliOrInner == DwsStrConstant.ins().num4){
            topic = "dws_multidimensional_obc_debug";
        }
        String groupId = topic.concat("#1");
        System.out.println(" 来源dwmTopic ===== "+topic);
        System.out.println(String.format(" groupId ===== [ kafka-consumer-groups --bootstrap-server %s --describe --group %s",kafkaServer,groupId));
        FlinkKafkaConsumer<String> consumer = FlinkKafkaUtil.getConsumer(kafkaServer, groupId, topic);
        if (aliOrInner== DwsStrConstant.ins().num1){
            consumer.setStartFromLatest();
        }else {
            consumer.setStartFromEarliest();
        }
        //2.字段映射表
        JSONObject srcJson = ChargePredictFields.ins().srcInfo;
        ChargePredictSrcStandardMapFunction srcStandard = new ChargePredictSrcStandardMapFunction(srcJson);
        //3.flink通过consumer读取kafka数据 同时进行字段标准化
        DataStream<JSONObject> sourceDataStream = env
                .addSource(consumer)
                .map(srcStandard)
                .returns(Types.GENERIC(JSONObject.class))
                ;
        if (true){
            //用于模拟es13实时数据
            DataStream<JSONObject> filter = odsEs13Source(sourceDataStream);
            return tempAndDcObc(filter);
        }
        DataStream<JSONObject> selector = tempAndDcObc(sourceDataStream);
        return selector;
    }

    private DataStream<JSONObject> odsEs13Source(DataStream<JSONObject> sourceDataStream) {
        DataStream<JSONObject> filter = sourceDataStream.filter((FilterFunction<JSONObject>) line -> {
            try {

                Float batter = line.getFloat(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
                String tid = line.getString(ChargePredictFields.ins().srcFieldTid);
                boolean obcIsOne = StringUtils.equalsIgnoreCase(line.getString(ChargePredictFields.ins().srcFieldBmsObcConnect), "1");
                boolean socNotContains99 = !StringUtils.containsIgnoreCase(line.getString(ChargePredictFields.ins().srcFieldBmsSoc), ChargePredictFields.ins().defaultVal);
                //充电组名称 如果为空或者-999999 则舍弃
                boolean startChargeTime = line.containsKey(ChargePredictFields.ins().srcFieldStartChargeTime);
                String chargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
                boolean startChargeTimeIsNotNull = !(StringUtils.isBlank(chargeTime) || StringUtils.equalsIgnoreCase(chargeTime, ChargePredictFields.ins().defaultVal));
                int timeCompare = tid.compareTo(chargeTime);
                boolean b =  startChargeTime && startChargeTimeIsNotNull && batter < 0 && socNotContains99 && timeCompare>=0 && obcIsOne;
                return b;
            }catch (Exception e ){
                e.printStackTrace();
                return false;
            }
        });
        if (false){
            filter = filter.filter(new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject value) throws Exception {
                    String vin = value.getString(ChargePredictFields.ins().srcFieldVin);
                    String startChargeTime = value.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
                    boolean b = StringUtils.equalsIgnoreCase(vin, "LGWEEUA52NE202588") && StringUtils.equalsIgnoreCase(startChargeTime, "1681297814000");
                    if (b){
                        System.out.println();
                        return true;
                    }
                    return false;
                }
            });
        }
        return filter;
    }
    /**
     * @description: 温度和充电方式的映射
     * @author: GW00280745
     * @date: 2022/5/11 13:51
     * @param: chargeDwmConf
     * @param: sourceDataStream
     * @return: DataStream<JSONObject>
     **/
    private DataStream<JSONObject> tempAndDcObc(DataStream<JSONObject> sourceDataStream) {
        DataStream<JSONObject> map = sourceDataStream
                .map((MapFunction<JSONObject, JSONObject>) line -> {
                    Integer tempInt = ChargePredictStrategy.ins()
                            .temp(line.getInteger(ChargePredictFields.ins().srcFieldBmsRmcModuleTempMax),
                                    line.getInteger(ChargePredictFields.ins().srcFieldBmsRmcModuleTempMin));
                    line.put(ChargePredictFields.ins().stdFieldTemp, tempInt);
                    Integer dcObc = ChargePredictStrategy.ins().dcObc(line.getIntValue(ChargePredictFields.ins().srcFieldBmsObcConnect),
                            line.getIntValue(ChargePredictFields.ins().srcFieldBmsDcConnect));
                    line.put(ChargePredictFields.ins().stdFieldDcObcSts, dcObc);
                    return line;
                });
        return map;
    }
    /**
     * @description: 动态配置信息--当前主要是读取web端配置,实时更新
     * @author:  GW00280745
     * @date: 2022/7/23 10:40
     **/
    private BroadcastStream<JSONObject> dynamicConfInfo(StreamExecutionEnvironment env) {
        String url = ChargePredictFields.ins().confInfo.getOrDefault("mysql.url","jdbc:mysql://10.255.128.243:3306/vp_business?useUnicode=true&characterEncoding=utf8&useSSL=false");
        String user = ChargePredictFields.ins().confInfo.getOrDefault("mysql.user","root");
        String password = ChargePredictFields.ins().confInfo.getOrDefault("mysql.password","vaas_sql*");
        String sql = ChargePredictFields.ins().confInfo.getOrDefault("mysql.sql","select topic,config_key,config_value from config_data where topic='chargePredict'");
        String timeout = ChargePredictFields.ins().confInfo.getOrDefault("mysql.interval_time","60");
        MapStateDescriptor<String, Map<String, Object>> confInfo = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));
        BroadcastStream<JSONObject> broadcastStream = env.addSource(new ChargePredictMysqlSourceFunction(url, user, password, sql, timeout))
                .setParallelism(1)
                .broadcast(confInfo)
                ;
        return broadcastStream;
    }

    /**
     * @description: 增量检查点，表示在 checkpoint 时，只备份和上个检查点相比，发生变化的检查点==增量更新
     * @author:  GW00280745
     * @date: 2022/7/1 15:34
     * @param:
     * @param: env
     * @return:
     **/
    private void checkPoint(StreamExecutionEnvironment env) {
        //1. 设置 state backend 状态后端
        //state多的时候需要开启增量检查点 如果是全量检查点并且状态数据较多那么会导致checkpoint失败
        //true代表增量检查点
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

        //2.checkpoint
        CheckpointConfig config = env.getCheckpointConfig();
        // 设置模式为exactly-once ; 模式支持EXACTLY_ONCE()/AT_LEAST_ONCE()
        config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的周期, 每隔5000 ms进行启动一个检查点 ; 启用 checkpoint,设置触发间隔（两次执行开始时间间隔）
        config.setCheckpointInterval(120*1000L);
        // 检查点必须在一分钟内完成，或者被丢弃 checkpoint的超时时间
        config.setCheckpointTimeout(60*1000L);
        // 同时并发数量; 同一时间只允许进行一个检查点
        config.setMaxConcurrentCheckpoints(1);
        // 任务流取消和故障时会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint ; 外部checkpoint(例如文件存储)清除策略
        config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //存储位置，FileSystemCheckpointStorage(文件存储)
        config.setCheckpointStorage(cpDir);
        // 确保检查点之间有至少500 ms的间隔-checkpoint最小间隔 ;最小间隔时间（前一次结束时间，与下一次开始时间间隔）
        config.setMinPauseBetweenCheckpoints(100*1000);
        //失败重试
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
        config.setTolerableCheckpointFailureNumber(CheckpointFailureManager.UNLIMITED_TOLERABLE_FAILURE_NUMBER);
    }
}
